package com.sl.core.engine.log;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.ReflectUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.sl.core.engine.listener.PipelineExecutionListener;
import com.sl.core.engine.trans.ProcessInstance;
import com.sl.core.utils.JsonUtil;
import lombok.SneakyThrows;
import org.apache.commons.compress.utils.Lists;
import org.apache.hop.core.logging.HopLoggingEvent;
import org.apache.hop.core.logging.LoggingBuffer;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class ProcessLoggingBuffer extends LoggingBuffer {

    private static ProcessLoggingBuffer INSTANCE;

    private Cache<String, ArrayBlockingQueue<HopLoggingEvent>> trackLog = CacheBuilder.newBuilder().maximumSize(5000).expireAfterWrite(1, TimeUnit.HOURS).build();

    public ProcessLoggingBuffer(int bufferSize) {
        super(bufferSize);
    }

    public static ProcessLoggingBuffer getInstance() {
        if (ObjectUtil.isNull(INSTANCE)) {
            INSTANCE = new ProcessLoggingBuffer(5000);
        }
        return INSTANCE;
    }

    @SneakyThrows
    @Override
    public void doAppend(HopLoggingEvent event) {

        if (ObjectUtil.isNull(event)) {
            return;
        }
        ProcessLoggingEvent processLoggingEvent = null;
        if (!(event instanceof ProcessLoggingEvent)) {
            processLoggingEvent = new ProcessLoggingEvent(event);
        } else {
            processLoggingEvent = (ProcessLoggingEvent) event;
        }
        String tid = processLoggingEvent.getTid();
        String instanceId = processLoggingEvent.geInstanceId();

//        //这里要给前端可视化的
//        String format = layout.formatJson(processLoggingEvent);

        Integer bufferSize = (Integer) ReflectUtil.getFieldValue(this, "bufferSize");

        String key = buildKey(tid, instanceId);

        ArrayBlockingQueue<HopLoggingEvent> kettleLoggingEvents = trackLog.get(key, () -> new ArrayBlockingQueue<>(bufferSize));
        try {
            boolean offer = kettleLoggingEvents.offer(event);
            //队列满了,先存储一把
            if (BooleanUtil.isFalse(offer)) {
                Boolean b = tempSaveDetail(processLoggingEvent);
                if (BooleanUtil.isFalse(b)) {
                    this.removeByKey(key);
                }
                doAppend(event);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public String buildKey(String tid, String instanceId) {
        return StrUtil.concat(true, tid, "@_@", instanceId);
    }

    public Boolean tempSaveDetail(ProcessLoggingEvent processLoggingEvent) {
        ProcessLogLayout layout = (ProcessLogLayout) ProcessLoggingBuffer.getInstance().getLayout();
        ProcessInstance pipeline = layout.getPipeline(processLoggingEvent);
        if (ObjectUtil.isNull(pipeline)) {
            return false;
        }
        PipelineExecutionListener.INSTANCE.saveDetail(pipeline);
        return true;
    }

    @Override
    public StringBuffer getBuffer(String key, boolean includeGeneral) {
        ArrayBlockingQueue<HopLoggingEvent> kettleLoggingEvents = trackLog.getIfPresent(key);
        if (CollUtil.isEmpty(kettleLoggingEvents)) {
            return new StringBuffer();
        }

        List<HopLoggingEvent> collect = kettleLoggingEvents.stream().sorted(Comparator.comparingLong(HopLoggingEvent::getTimeStamp)).collect(Collectors.toList());

        List<Map<String, Object>> maps = Lists.newArrayList();
        for (HopLoggingEvent event : collect) {
            ProcessLogLayout layout = (ProcessLogLayout) getLayout();

            ProcessLoggingEvent processLoggingEvent = null;
            if (!(event instanceof ProcessLoggingEvent)) {
                processLoggingEvent = new ProcessLoggingEvent(event);
            } else {
                processLoggingEvent = (ProcessLoggingEvent) event;
            }
            Map<String, Object> map = layout.formatJsonToMap(processLoggingEvent);
            maps.add(map);

        }
        String str = null;
        try{
             str = JsonUtil.toStr(maps);
        }catch (Exception e){
            e.printStackTrace();
        }
        if(StrUtil.isBlank(str)){
            str="";
        }
        return new StringBuffer(str);
    }

    public void removeByKey(String key) {
        trackLog.invalidate(key);
    }

    public String pollAllByKey(String key) {
        StringBuffer buffer = this.getBuffer(key, true);
        this.removeByKey(key);
        return buffer.toString();
    }
}
